package com.bfxy.kafka.api.serial;

import com.bfxy.kafka.api.model.User;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class UserSerializer implements Serializer<User> {

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public byte[] serialize(String s, User user) {
        if (null == user) {
            return null;
        }
        byte[] idBytes, nameBytes;
        try {
            String id = user.getId();
            String name = user.getName();
            if (null != id) {
                idBytes = id.getBytes(StandardCharsets.UTF_8);
            } else {
                idBytes = new byte[0];
            }
            if (null != name) {
                nameBytes = name.getBytes(StandardCharsets.UTF_8);
            } else {
                nameBytes = new byte[0];
            }
            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + idBytes.length + nameBytes.length);
            // 4个字节也就是一个int类型长度：这个putInt(int value)方法放idBytes的实际真实长度
            buffer.putInt(idBytes.length);
            // put(byte[] src)实际放的是idBytes真实的字节数组，也就是内容
            buffer.put(idBytes);
            // 4个字节也就是一个int类型长度：这个putInt(int value)方法放nameBytes的实际真实长度
            buffer.putInt(nameBytes.length);
            // put(byte[] src)实际放的是nameBytes真实的字节数组，也就是内容
            buffer.put(nameBytes);
            return buffer.array();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public void close() {

    }

}
